package com.pai4j.ugc.messaegqueue.consumer.rocketmq;

import com.pai4j.common.util.JsonUtil;
import com.pai4j.domain.dto.queue.MomentPublishMessageDTO;
import com.pai4j.ugc.messaegqueue.service.MomentPublishConsumeService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


/**
 * 动态发布消息RocketMQ消费者
 * 负责处理动态发布后的后置处理任务
 *
 * @author: CYM-pai
 * @date: 2025/01/04 12:00
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "pai4j_ugc", topic = "queue_moment_publish")
public class MomentPublishRocketMQConsumer implements RocketMQListener<String> {

    @Autowired
    private MomentPublishConsumeService momentPublishConsumeService;

    /**
     * 消费动态发布消息
     * 执行动态发布后的后置处理任务
     *
     * @param message 消息内容
     *
     * @author: CYM-pai
     * @date: 2025/01/04 12:05
     */
    @Override
    public void onMessage(String message) {
        try {
            log.info("接收到动态发布消息: {}", message);

            // 反序列化消息
            MomentPublishMessageDTO data = JsonUtil.fromJson(message, MomentPublishMessageDTO.class);
            if (data == null || data.getMoment() == null) {
                log.warn("动态发布消息格式错误或为空: {}", message);
                return;
            }

            // 执行动态发布后置处理
            momentPublishConsumeService.consume(data.getMoment());

            log.info("动态发布消息处理完成: momentId={}, author={}",
                    data.getMoment().getId(), data.getMoment().getAuthor());

        } catch (Exception e) {
            log.error("处理动态发布消息失败: {}", message, e);
            // 这里可以根据业务需求决定是否重试或者记录到死信队列
            try {
                throw e; // 抛出异常让RocketMQ进行重试
            } catch (Exception ex) {
                throw new RuntimeException(ex);
            }
        }
    }
}